package jmine.tec.cluster.heartbeat;

import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Map.Entry;

import jmine.tec.cluster.MasterIdentificator;
import jmine.tec.cluster.dao.ClusterNodeDAO;
import jmine.tec.cluster.domain.ClusterNode;

import org.hibernate.LockMode;
import org.hibernate.StaleStateException;
import org.hibernate.UnresolvableObjectException;

/**
 * Implementação de identificador de nó mestre, em que o nó que abrigar o clusterBean de menor id é o mestre.
 * 
 * @author ricardo
 */
public class HeartBeatMasterIdentificator implements MasterIdentificator {

    private ClusterNodeDAO clusterNodeDAO;

    private volatile Map<Long, Long> allNodes = new LinkedHashMap<Long, Long>();

    private volatile ClusterNode node;

    /**
     * Construtor
     * 
     * @param clusterNodeDAO o dao
     */
    public HeartBeatMasterIdentificator(ClusterNodeDAO clusterNodeDAO) {
        this.clusterNodeDAO = clusterNodeDAO;
    }

    /**
     * {@inheritDoc}
     */
    public String getHostname() {
        try {
            return InetAddress.getLocalHost().getHostName();
        } catch (UnknownHostException e) {
            return "UNKNOWN_HOST";
        }
    }

    /**
     * {@inheritDoc}
     */
    public boolean isMaster() {
        ClusterNode currentNode = this.node;
        if (currentNode == null) {
            return false;
        } else if (currentNode.getId() == null) {
            return false;
        }
        return this.clusterNodeDAO.isMaster(currentNode.getId());
    }

    /**
     * {@inheritDoc}
     */
    public void validateNodes() {
        Map<Long, Long> oldClusterNodes = this.allNodes;
        Map<Long, ClusterNode> newClusterNodes = new LinkedHashMap<Long, ClusterNode>(oldClusterNodes.size());

        for (ClusterNode n : this.clusterNodeDAO.findAllOrderedByPk()) {
            newClusterNodes.put(n.getId(), n);
        }
        for (Entry<Long, Long> entry : oldClusterNodes.entrySet()) {
            Long id = entry.getKey();
            Long oldPingValue = entry.getValue();
            ClusterNode newStateNode = newClusterNodes.get(id);
            if ((newStateNode != null) && this.isNodeStale(id, oldPingValue, newStateNode)) {
                try {
                    this.clusterNodeDAO.getSession().lock(newStateNode, LockMode.UPGRADE);
                    newStateNode.getPersister().remove();
                    newClusterNodes.remove(id);
                } catch (UnresolvableObjectException e) {
                    // OK, another node cleaned it first
                }
            }
        }
        Map<Long, Long> nextAllNodes = new LinkedHashMap<Long, Long>();
        for (Entry<Long, ClusterNode> entry : newClusterNodes.entrySet()) {
            nextAllNodes.put(entry.getKey(), entry.getValue().getPing());
        }
        this.allNodes = nextAllNodes;
    }

    /**
     * Verifica se o nó está ativo ou inativo. Se for inativo, remover o bean newStateNode do banco de dados e deixar nula a referência ao
     * ClusterNode interno desta classe.
     * 
     * @param id o identificador do bean do nó que será verificado
     * @param oldPingValue o contador de atualização anterior
     * @param newStateNode o bean mais atual
     * @return true se newStateNode está inativo; false, caso contrário
     */
    private boolean isNodeStale(Long id, Long oldPingValue, ClusterNode newStateNode) {
        if (oldPingValue.equals(newStateNode.getPing())) {
            try {
                this.clusterNodeDAO.getSession().lock(newStateNode, LockMode.UPGRADE);
            } catch (StaleStateException e) {
                // nada a fazer, algum outro nó já apagou o nó parado.
            }
            return true;
        }

        return false;
    }

    /**
     * {@inheritDoc}
     */
    public synchronized void heartBeat() throws Exception {
        ClusterNode currentNode = this.node;
        if (currentNode == null) {
            currentNode = this.clusterNodeDAO.createBean();
            currentNode.setHostname(this.getHostname());
            currentNode.getPersister().save();
            this.node = currentNode;
        } else {
            try {
                this.clusterNodeDAO.getSession().refresh(currentNode, LockMode.UPGRADE);
                currentNode.setPing(currentNode.getPing() + 1);
                currentNode.getPersister().save();
                this.node = currentNode;
            } catch (UnresolvableObjectException e) {
                this.node = null;
                this.heartBeat();
                return;
            }
        }
    }

    /**
     * @return o nó do cluster.
     */
    public ClusterNode getNode() {
        return this.node;
    }

    /**
     * {@inheritDoc}
     */
    public int getTotalNodes() {
        return this.allNodes.size();
    }

    /**
     * Método de inicialização a ser chamado pelo spring
     * 
     * @throws Exception ex
     */
    public void initializeNode() throws Exception {
        this.heartBeat();
    }
}